package com.ruins.ping.producer

import com.alibaba.fastjson.JSONObject
import com.ruins.ping.PingApplication
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.test.context.ContextConfiguration
import org.springframework.util.concurrent.ListenableFuture
import spock.lang.Specification
import spock.lang.Unroll

import java.lang.management.ManagementFactory
import java.lang.management.RuntimeMXBean

import static org.mockito.Mockito.mock

@Unroll
@ContextConfiguration(classes = PingApplication.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class LogSenderTest extends Specification {

    LogSender logSender
    KafkaTemplate<String, String> kafkaTemplate = Mock(KafkaTemplate)
    ListenableFuture<String> future = mock(ListenableFuture.class);

    def setup() {
        kafkaTemplate.send(_ as String,_) >> future

        logSender = new LogSender(kafkaTemplate)
    }

    def "should send log message to Kafka"() {
        given:
        String logMessage = "This is a test log message"
        String expectedPid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0]

        when:
        logSender.sendLog(logMessage)

        then:
        1 * kafkaTemplate.send("logTopic", { data ->
            JSONObject parsedData = JSONObject.parseObject(data)
            parsedData.getString("logMessage") == logMessage &&
                    parsedData.getString("pid") == expectedPid
        })
    }
}